查看原文
其他

解决并发编程之痛的良药--结构化并发编程

曹家锋 高可用架构 2020-11-06
作者简介:曹家锋,Westar实验室技术专家。Westar实验室(westar.io),成立于 2018 年,关注于区块链及分布式前沿技术,包括区块链分层架构、二层路由,网络性能、智能合约、PoW 优化等。


并发,是程序员在日常编程中难以绕开的话题,本文介绍一种并发编程范式-结构化并发(Structured Concurrency)。首先给出它的概念和现状,然后着重介绍 Rust 的一个实现 - task_scope,最后给出一个例子展示如何在实践中使用。


The Pain of concurrency programming


熟悉 Go 语言的朋友都知道,可以通过 go myfunc() 轻易的创建一个和当前协程并发执行的 task。但是,当程序变复杂, go statement 变的越来越多时,就会遇到各种 task 生命周期的问题。

  • 这个 task 什么时候开始,什么时候结束?

  • 怎么做到当所有 subtask 都结束,main task再结束?

  • 假如某个 subtask 失败,main task 如何cancel 掉其他subtask?

  • 如何保证所有 subtask 在某个特定的超时时间内返回,无论它成功还是失败?

  • 更进一步,如何保证 main task 在规定的时间内返回,无论其成功还是失败,同时 cancel 掉它产生的所有 subtask?

  • main task 已经结束了,subtask 还在 running,是不是存在资源泄漏?


以上只是我根据自己过往的经验,随便列举的几类问题。当然这些问题在 Golang 里面都是可以解的,具体可以参考 Golang Official Blog 里几篇讲 Golang Concurrency Patterns 的文章。它需要程序按照一些特定的行为方式去组织,比如说方法参数带上 Context,通过它去传递 cancellation 信号。

Go Concurrency Patterns: Context[1]

Go Concurrency Patterns: Pipelines and cancellation[2]

Go Concurrency Patterns: Timing out, moving on[3]

Advanced Go Concurrency Patterns[4]

在多线程模型中,上面几个问题给程序员带来了更多复杂性和更重的心智负担。我相信大部分 Java 程序员都无法准确的把上面几个问题都解决掉,这不是嘲讽,而是线程模型本身给使用者带来的诸多问题,这对使用者的要求实在是太高了。

那么,有没有一种编程范式,既可以解决这些问题,又具有相对比较低的认知门槛,同时也不需要像 Golang Context 那样侵入应用程序的接口?结构化并发(Structured Concurrency) 就是这样一种并发编程范式。


Structured Concurrency


2016年,ZerMQ 的作者 Martin Sústrik 在他的文章[5]中第一次形式化的提出结构化并发这个概念。2018 年 Nathaniel J. Smith (njs) 在 Python 中实现了这一范式 - trio[6],并在 Notes on structured concurrency, or: Go statement considered harmful[7] 一文中进一步阐述了 Structured Concurrency。同时期,Roman Elizarov 也提出了相同的理念[8],并在 Kotlin 中实现了大家熟知的kotlinx.coroutine[9]。2019年,OpenJDK loom project 也开始引入 structured concurrency,作为其轻量级线程和协程的一部分。


废话这么多,一方面是想提供更多的历史,方便读者更深入的了解,另一方面也是想说明,结构化并发虽然还是一个比较新的概念,具体的细节也在不断演进中,但已经有成熟的工业界实现,读者可以在自己熟悉的语言中应用该范式。

lidill(C): http://libdill.org/

trio(Python): https://trio.readthedocs.io/en/stable/

kotin.coroutine: https://github.com/Kotlin/kotlinx.coroutines

Venice(Swift): https://github.com/Zewo/Venice

Structured Concurrency 核心在于通过一种 structured 的方法实现并发程序,用具有明确入口点和出口点的控制流结构来封装并发“线程”(可以是系统级线程也可以是用户级线程,也就是协程,甚至可以是进程)的执行,确保所有派生“线程”在出口之前完成。


说的可能有点抽象,举个例子。


func main_func() { go myfunc() go anotherfunc() <rest of program>...}


假设上图中的代码具有 structured concurrency 特性(这里用的是 golang 的语法来展示)。main_func 里,创建了两个子任务:myfunc(), anotherfunc,这里的 func 是一个控制流结构,入口就是 func 调用开始,出口是 func 调用结束,派生出来的两个子任务需要在 main_func 调用结束之前先完成。当 main_func 结束,它涉及到的资源也都会被释放掉。外部调用者无法也无需感知 main_func 里面到底是串行的还是并行的,它只需要调用 main_func,然后等待它结束即可。这就是所谓的 Structured。

大家应该都知道 goto 语句,一般不推荐使用它(见Dijkstra: Go To Statement Considered Harmful),使用 goto 的场景基本都可以用 if, else, for loop, while loop 这些控制结构组合表达,可以把这些控制结构叫做 structured statement。

Structured Concurrency 的概念和 structured statement 类似,通过控制流来保证并发语义的可控,而不是 go coroutine 满天飞。

关于这方面的类比,njs 在 Notes on structured concurrency, or: Go statement considered harmful 中做了详细的说明,推荐阅读。

以上是 Structured Concurrency 的核心概念,看起来是不是很简单。下面就跟着我去看看,在 Rust 里你可以怎样实现 Structured Concurrency。


Implement Structured Concurrency in Rust


目前 Rust 并没有一个成熟的库支持 Structured Concurrency 的编程范式。但是 tokio#1879[10] 这个 issue 中已经开始讨论了如何在 tokio 这个底层库中提供支持,以实现 structured concurrency 风格的编程。如果你比较感兴趣,欢迎去这里贡献你的力量。


本节以 Rust 社区另外一个库 - task_scope 来介绍这种编程范式。task_scope 是一个日本小哥写的一个试验性质的库。在阅读和试验它时,我认为它提供的接口在使用上很别扭,不便于实现更复杂的并发逻辑,于是基于自己的经验,我把它的对外接口抽象成 Scope 和 CancelScope。这两个概念是继承自 trio 的实现,Scope 对应 trio 的 nursery,CancelScope 对应 trio 的 cancel_token。fork 版本见 startcoinorg/task_scope[11]。


Scope


为了将具有 Structured Concurrency 行为的代码与普通的异步代码区别出来,我在 task_scope 中引入了Scope 这个实体。所有 structured concurrency 的异步代码都必须在 Scope 的作用域中完成。下面给出用 task_scope 实现之前例子的伪代码。


let scope = Scope::new();scope.run(|spawner| async { spawner.spawn(myfunc()); spawner.spawn(anotherfunc()); <rest of program>...}).await;


Scope 作为 Structured Concurrency 的控制结构,任何想要进行 structured concurrency 编程的代码都必须初始化出一个 Scope 对象,调用 Scope.run 打开了这个控制结构的入口,在控制结构里面,可以随意的 spawn 子任务。myfunc 和 anotherfunc 都是运行在这个 scope 里。没有 Scope,父任务无法开启新的子任务,这保证了 Scope 是 Structured Concurrency 的唯一入口。最重要的是,只有当所有子任务都结束时,父任务才会结束,如果父任务在子任务结束前,就已经执行完自己的代码块,那么它需要暂停自己,并等待所有子任务结束。


Golang 的 go 语句最主要的问题是,当你调用了一个函数,并且函数返回了,然而你不知道它是否开启了一个/些后台任务,这些后台任务在函数返回后也不会结束(无论是有意的还是无意的)。这打破了函数的抽象,破坏了它的封装性。通过 Scope 抽象的 Structured Concurrency,就没有这个问题。任何一个函数都可以通过 scope 来运行多个并发的任务,但是函数无法返回,除非所有的子任务都完成了。因此,当一个函数返回了,你知道它是真的返回了,而不会有其他遗漏的子任务。


Timeout and Cancellation


还记得我们在开篇提到的几个问题吗,里面涉及到超时和取消:如何保证 main task 在规定的时间内返回,无论其成功还是失败,同时 cancel 掉它产生的所有 subtask? 这一节,我们来聊聊这个问题。


我在 task_scope 中为 Scope 提供了一个方法 pub fn cancel_scope(&self) -> CancelScope,来获取这个 scope 的 cancel_token。调用 CancelScope.cancel/force_cancel 方法可以取消正在执行的 scope,前者给予 scope 一定的机会做优雅退出,后者则没有。以下是一个更加完善的例子,加入 cancel scope 的概念。


let scope = Scope::new();let cancel_token = scope.cancel_scope();scope.run(|spawner| async { let handle1 = spawner.spawn(myfunc()); let handle2 = spawner.spawn(anotherfunc()); <rest of program>... let result = select(handle1, handle2, delay(1000)).await; if let Err(Timeout) = result { cancel_token.cancel(); }}).await;


main task 的最后,加入一个超时判断,select(handle1, handle2, delay(1000)).await,如果 handle1 和handle2 都没有在 delay(1000) 之后完成,那么就返回 Timeout,然后通过 cancel_token.cancel()取消scope的执行,这会导致 scope 里所有 child tasks 都收到 Cancel 信号,这些 child task 在下一次被调度器调度执行时,会直接退出执行。(task_scope 无法打断正在被调度器执行的 future,所以只能等到 future yield 后,下次被调度时退出,也就是说,future 中 await 的地方就是 cancel 信号的 checkpoint)。


Scope in Scope


当并发逻辑变得复杂,我们就会遇到在 Scope 里面开启新 Scope 的情况。一般来讲, scope 会被封装在函数里,函数的外部调用者不知道函数里是否开启了 scope。假如说,外部调用者本身是在一个 scope 里调用这个函数,就会出现 scope in scope。这种情况下,Structured Concurrency 的特性依然保持不变。


let scope = Scope::new();scope.run(|spawner| async { spawner.spawn(func_a()); spawner.spawn(anotherfunc()); <rest of program>...}).await;
async fn func_a() { let scope = Scope::new(); scope.run(|spawner| async { spawner.spawn(myfunc()); <rest of program>... }).await}


func_a 是一个封装有 scope 的函数,当它的子任务都完成时,它才会返回。外部调用 spawn 了 func_a 作为子任务执行,它也会等 func_a 完成后再结束。


timeout 和 cancellation 又如何处理的呢?当外部调用者的 scope 被 cancel 时, cancel 信号传递到每个 child task 里,child task future 检查自己是否被外部的 scope cancel 掉。


  • 如果是 graceful cancel,它会给自己的子任务也发送 graceful cancel 信号,然后继续执行或者等待,直到所有子任务都退出;

  • 如果是 force cancel,那它就给自己的子任务发送 force cancel ,然后直接退出。


这样,cancel 信号,就会通过 scope -> subscope -> sub-subscope 一层一层的往下传递,形成一个 cancel tree,通过 root 往下派发。


Practice


讲完这些基本概念,最后给读者留一个比较经典的练习题去尝试下使用 Structured Concurrency 编程。


Happy Eyeballs[12] 算法是 njs 在演示 trio 时使用的示例,他给出的 Python 实现[13]可以在这里找到。这是一个非常好的示例,强烈建议读者动手去试一试如何利用自己已有的经验去实现它(很有可能你写不出来),然后再尝试用 Strucutred Concurrency 的范式去实现。


最后,给出我用 task_scope 实现的 Rust 版本(https://github.com/starcoinorg/task_scope/blob/master/examples/happy_eyeball.rs)。


文中链接

[1] https://blog.golang.org/context

[2] https://blog.golang.org/pipelines

[3] https://blog.golang.org/go-concurrency-patterns-timing-out-and

[4] https://blog.golang.org/advanced-go-concurrency-patterns

[5] http://250bpm.com/blog:71

[6] https://trio.readthedocs.io/en/stable/

[7] https://vorpus.org/blog/notes-on-structured-concurrency-or-go-statement-considered-harmful/

[8] https://medium.com/@elizarov/structured-concurrency-722d765aa952

[9] https://github.com/Kotlin/kotlinx.coroutines

[10] https://github.com/tokio-rs/tokio/issues/1879

[11] https://github.com/starcoinorg/task_scope

[12] https://en.wikipedia.org/wiki/Happy_Eyeballs

[13] https://github.com/python-trio/trio/blob/master/trio/_highlevel_open_tcp_stream.py


参考阅读:


技术原创及架构实践文章,欢迎通过公众号菜单「联系我们」进行投稿。


高可用架构

改变互联网的构建方式


长按二维码 关注「高可用架构」公众号


    您可能也对以下帖子感兴趣

    文章有问题?点此查看未经处理的缓存